package com.atguigu.api

import org.apache.flink.streaming.api.functions.source.SourceFunction

import scala.util.Random

/**
 * @description: 自定义数据源重点,可以生成自定义数据源测试数据
 * @time: 2020/6/19 16:05 封装,继承,多态,抽象 模式匹配,柯里化,惰性求值,闭包
 * @author: baojinlong
 **/
class MySensorSource extends SourceFunction[SensorReading] {
  // 定义一个flag表示数据源是否正常运行
  var running: Boolean = true

  /**
   * 随机生成SensorReading数据
   *
   * @param sourceContext
   */
  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    val rand = new Random
    // 随机生成10个传感器的温度值,并且不停在之前基础上更新(随机上下波动)
    val inclusive: Range.Inclusive = 1.to(10)
    val tuples: Seq[(String, Double)] = inclusive.map(x => ("sensor_" + x, 60 + rand.nextGaussian() * 20))
    while (running) {
      val curTemps: Seq[(String, Double)] = tuples.map(data => {
        (data._1, data._2 + rand.nextGaussian())
      })
      // 获取当前系统时时间
      val curTs: Long = System.currentTimeMillis
      curTemps.foreach(data => sourceContext.collect(SensorReading(data._1, curTs, data._2)))
      // 定义间隔时间
      Thread.sleep(1000L)
    }

  }

  override def cancel(): Unit = running = false
}
